/*
 ----------------------------------------------------------------------------------
                           "THE BEER-WARE COPYRIGHT LICENSE"
                                     (Revision 43)
                                      -----------
    Rymek [rymek@rymek.org] wrote this file. As long as You retain this notice and
    this file has been sent directly to You by the author, You can do whatever You
    want with this stuff as long as You don't sell or rent it. If we meet some day,
    and You think this stuff is worth it, You can buy me a beer in return :-)
                                                                 __________
                                                                 Paweł Rymaszewski
  ----------------------------------------------------------------------------------
 */

package processing.logic;

import constants.MessageConstants;
import processing.logic.threads.FetcherThread;
import processing.logic.util.Connector;
import processing.objects.Result;
import processing.objects.Results;
import processing.objects.Server;
import utils.properties.Configuration;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Observable;
import java.util.Observer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Class for fetching body of the messages
 *
 * @author rymek
 *         User: rymek
 *         Date: 30.06.11
 *         Time: 10:10
 */
public class Fetcher extends Observable implements Observer {

    // thread pool for database connections
    private final ExecutorService threadPool =
            Executors.newFixedThreadPool(Configuration.generalProperties.getDatabaseThreadLimit());

    // list of all worker threads
    private final ArrayList<Connector> threadList = new ArrayList<>();

    private final Server server;

    /**
     * Constructor.
     * Also creates Datbase connection and prepares Thread Pool
     *
     * @param server server information
     */
    public Fetcher(Server server) {
        this.server = server;
    }

    /**
     * Method for fetching body of the messages
     *
     * @param unparsedResults results to be updated with message body and file paths
     * @param parsedResults   Result object to write parsed results into
     */
    public void fetchMessages(Results unparsedResults, Results parsedResults) {

        parsedResults.setOutputDir(this.server.getTnsName() + "_" + unparsedResults.getSearchParameters().getMessageInfo().getRequestName() + "_" + unparsedResults.getExecutionTime());

        for (Result unparsedResult : unparsedResults.getResults().get(MessageConstants.MSG_REQUEST).values()) {

            FetcherThread fetcherThread = new FetcherThread(unparsedResult, parsedResults, this.server);
            fetcherThread.addObserver(this);

            this.threadPool.submit(fetcherThread);
            this.threadList.add(fetcherThread);
        }
    }

    @Override
    /**
     * Observes created threads if they are finished
     */
    public void update(Observable o, Object arg) {
        super.setChanged();
        super.notifyObservers();
    }

    /**
     * Terminates all worker threads
     *
     * @throws SQLException when database access error occurs
     */
    public void terminateThreads() throws SQLException {

        // shuts down thread pool
        this.threadPool.shutdownNow();

        // terminates connections for worker threads
        for (Connector connector : this.threadList) connector.cleanup();
    }
}
